package com.zb.activemq;


import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.ScheduledMessage;

import javax.jms.*;
import java.util.Date;

/**
 * Created by dongyuncong on 2017/8/11.
 */
public class ActiveMQUtil {


    public static void deleteAllScheduleMessage(final ConnectionFactory connectionFactory) throws JMSException {
        Connection conn = connectionFactory.createConnection();
        Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
        Destination management = session.createTopic(ScheduledMessage.AMQ_SCHEDULER_MANAGEMENT_DESTINATION);
        MessageProducer producer = session.createProducer(management);
        Message request = session.createMessage();
        request.setStringProperty(ScheduledMessage.AMQ_SCHEDULER_ACTION, ScheduledMessage.AMQ_SCHEDULER_ACTION_REMOVEALL);
        producer.send(request);
        close(conn);
    }

    /*删除指定时间段内要发送的延时消息，单位 ms*/
    public static void deleteExpiredScheduleMessage(final ConnectionFactory connectionFactory, long start, long end) throws JMSException {
        Connection conn = connectionFactory.createConnection();
        Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
        Destination management = session.createTopic(ScheduledMessage.AMQ_SCHEDULER_MANAGEMENT_DESTINATION);
        MessageProducer producer = session.createProducer(management);
        Message request = session.createMessage();
        request.setStringProperty(ScheduledMessage.AMQ_SCHEDULER_ACTION, ScheduledMessage.AMQ_SCHEDULER_ACTION_REMOVEALL);
        request.setStringProperty(ScheduledMessage.AMQ_SCHEDULER_ACTION_START_TIME, Long.toString(start));
        request.setStringProperty(ScheduledMessage.AMQ_SCHEDULER_ACTION_END_TIME, Long.toString(end));
        producer.send(request);
        close(conn);
    }

    public static void deleteScheduleByMsgId(final ConnectionFactory connectionFactory, String msgId) throws JMSException {
        Connection conn = connectionFactory.createConnection();
        Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
        Destination management = session.createTopic(ScheduledMessage.AMQ_SCHEDULER_MANAGEMENT_DESTINATION);
        MessageProducer producer = session.createProducer(management);
        Message request = session.createMessage();
        request.setStringProperty(ScheduledMessage.AMQ_SCHEDULER_ACTION, ScheduledMessage.AMQ_SCHEDULER_ACTION_REMOVE);
        request.setStringProperty(ScheduledMessage.AMQ_SCHEDULED_ID, msgId);
        producer.send(request);
        close(conn);
    }

    private static void close(Connection connection) {
        if (null != connection)
            try {
                connection.close();
            } catch (Exception e) {
                System.out.println(e);
            }
    }

    public static void main(String[] args) throws Exception {
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnectionFactory.DEFAULT_USER,
                ActiveMQConnectionFactory.DEFAULT_PASSWORD,
                "tcp://39.108.78.245:61616");
//        deleteAllScheduleMessage(connectionFactory);
        deleteExpiredScheduleMessage(connectionFactory, new Date().getTime(), new Date().getTime() + 600 * 1000L);
    }
}
